package com.zetyun.tiger.connector.manage;


import com.fasterxml.jackson.annotation.JsonPropertyDescription;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.meta.jsonschema.annotation.JsonSchemaTitle;
import com.zetyun.rt.sdk.action.SinkAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisSentinelPool;

import java.util.HashSet;
import java.util.Objects;
import java.util.Set;

@ActionMeta(
        id = "tiger.redis-mana-cust-acct",
        tags = {"redis"},
        category = "sink",
        name = "HisManaCustAcctSink",
        description = "加载历史客户经理客户账号关系",
        parameterClass = ManageCustAcctParameter.class,
        status = ActionMeta.Status.PUBLIC
)
@Slf4j
public class ManageCustAcctSinkAction extends SinkAction {

    private ObjectMapper om;

    private ManageCustAcctParameter param;

    private SimpleRedisContainer rc;

    private String additionalKey;

    @Override
    public void init(OperatorContext context) {
        super.init(context);


        om = new ObjectMapper();
        param = context.getParameters();
        additionalKey = param.getAdditionalKey();
        FlinkJedisSentinelConfig conf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(param.getMasterName())
                        .setSentinels(param.getSentinels())
                        .setPassword(param.getPassword())
                        .setDatabase(param.getDatabase())
                        .build();
        rc = build(conf);
        try {
            rc.open();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void apply(RtEvent in) throws Exception {
        String conntrNo = in.getField(param.getManaNoField(), String.class).trim();
        String custNo = in.getField(param.getCustNoField(), String.class).trim();
        String acctNo = in.getField(param.getAcctNoField(), String.class).trim();

        String cacheName = additionalKey.concat(conntrNo);
        Set<String> accts = strToSet(rc.hget(cacheName, custNo));
        accts.add(acctNo);

        rc.hset(cacheName, custNo, setToStr(accts), null);
    }

    @Override
    public void close() throws Exception {
        super.close();
        rc.close();
    }

    @SneakyThrows
    private Set<String> strToSet(String jsonStr) {
        if (jsonStr == null || jsonStr.isEmpty()) {
            return Sets.newHashSet();
        }
        return om.readValue(jsonStr, new TypeReference<Set<String>>() {
        });
    }

    @SneakyThrows
    private String setToStr(Set<String> set) {
        return om.writeValueAsString(set);
    }


    private static SimpleRedisContainer build(FlinkJedisSentinelConfig jedisSentinelConfig) {
        Objects.requireNonNull(jedisSentinelConfig, "Redis sentinel config should not be Null");

        GenericObjectPoolConfig genericObjectPoolConfig = getGenericObjectPoolConfig(jedisSentinelConfig);

        JedisSentinelPool jedisSentinelPool = new JedisSentinelPool(jedisSentinelConfig.getMasterName(),
                jedisSentinelConfig.getSentinels(), genericObjectPoolConfig,
                jedisSentinelConfig.getConnectionTimeout(), jedisSentinelConfig.getSoTimeout(),
                jedisSentinelConfig.getPassword(), jedisSentinelConfig.getDatabase());
        return new SimpleRedisContainer(jedisSentinelPool);
    }

    private static GenericObjectPoolConfig getGenericObjectPoolConfig(FlinkJedisConfigBase jedisConfig) {
        GenericObjectPoolConfig genericObjectPoolConfig = jedisConfig.getTestWhileIdle() ? new JedisPoolConfig() : new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(jedisConfig.getMaxIdle());
        genericObjectPoolConfig.setMaxTotal(jedisConfig.getMaxTotal());
        genericObjectPoolConfig.setMinIdle(jedisConfig.getMinIdle());
        genericObjectPoolConfig.setTestOnBorrow(jedisConfig.getTestOnBorrow());
        genericObjectPoolConfig.setTestOnReturn(jedisConfig.getTestOnReturn());

        return genericObjectPoolConfig;
    }
}
